package org.jerrymouse.jsa4j.db.kv.cassandra;

import java.io.UnsupportedEncodingException;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.thrift.Cassandra.Client;
import org.apache.thrift.TException;
import org.jerrymouse.jsa4j.db.kv.DB;

public class CassandraDB implements DB {
	private static final String UTF8 = "UTF-8";
	private Cassandra.Client client;
	private static String KS = "Keyspace1";
	private static String CF = "Standard1";
	private static String CN = "cn";

	public void setClient(Client client) {
		this.client = client;
	}

	@Override
	public String get(String key) {
		Column col;
		String v = null;

		try {
			col = client.get(KS, key, getColumPath(), ConsistencyLevel.ONE)
					.getColumn();
			if (col == null)
				return null;
			v = new String(col.getValue(), "UTF-8");
		} catch (UnsupportedEncodingException e1) {
			e1.printStackTrace();
		} catch (InvalidRequestException e1) {
			e1.printStackTrace();
		} catch (NotFoundException e1) {
			return v;
		} catch (UnavailableException e1) {
			e1.printStackTrace();
		} catch (TimedOutException e1) {
			e1.printStackTrace();
		} catch (TException e1) {
			e1.printStackTrace();
		}
		return v;
	}

	@Override
	public String put(String key, String value) {
		if (key == null)
			return null;
		String valueOrg = get(key);
		if (valueOrg == null && value != null) {
			return insert(key, value);
		} else if (valueOrg != null && value == null) {
			return delete(key, value);
		} else if (valueOrg != null && value != null) {
			return update(key, value);
		}
		return null;
	}

	private String update(String key, String value) {
		return insert(key, value);
	}

	private String delete(String key, String value) {
		long timestamp = System.currentTimeMillis();
		try {
			client.remove(KS, key, getColumPath(), timestamp,
					ConsistencyLevel.ONE);
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		} catch (InvalidRequestException e) {
			e.printStackTrace();
		} catch (UnavailableException e) {
			e.printStackTrace();
		} catch (TimedOutException e) {
			e.printStackTrace();
		} catch (TException e) {
			e.printStackTrace();
		}
		return key;

	}

	private ColumnPath getColumPath() throws UnsupportedEncodingException {
		ColumnPath colPathName = new ColumnPath(CF);
		colPathName.setColumn(CN.getBytes(UTF8));
		return colPathName;
	}

	private String insert(String key, String value) {
		try {
			long timestamp = System.currentTimeMillis();
			client.insert(KS, key, getColumPath(), value.getBytes(UTF8),
					timestamp, ConsistencyLevel.ONE);
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		} catch (InvalidRequestException e) {
			e.printStackTrace();
		} catch (UnavailableException e) {
			e.printStackTrace();
		} catch (TimedOutException e) {
			e.printStackTrace();
		} catch (TException e) {
			e.printStackTrace();
		}
		return key;
	}
}
